// Copyright (c) 2023 Presto Labs Pte. Ltd.
// Author: fengyang

#include "coin2/exchange/maicoin/feed_v2/processor.h"

#include <string>
#include <utility>

#include "coin2/base/conversion.h"
#include "coin2/exchange/base/api_util/json_schema.h"
#include "coin2/exchange/maicoin/feed_v2/update.h"
#include "coin2/exchange/util/json_util.h"
#include "coin2/exchange/maicoin/symbology/product.h"

namespace coin2::exchange::maicoin::feed_v2::impl {

void MaicoinFeedProcessor::OnTopicRecord(
    const base::executor::TopicRecordPacket& packet) {
  upd_mgr_.onBeginPacket(packet);

  rapidjson::Document document;
  document.Parse<rapidjson::kParseNumbersAsStringsFlag>(packet.payload.data());
  if (document.HasParseError()) {
    throw std::runtime_error(std::string("JSON parse error: ") +
                             GetParseError_En(document.GetParseError()));
  }

  if (parser_config_.validate_schema() && schema_) {
    schema_->AcceptOrThrow(document);
  }

  std::string channel;
  if (document.HasMember("c")) {
    channel = document["c"].GetString();
  } else if (document.HasMember("url")) {
    const std::string url = document["url"].GetString();
    if (url.find("/api/v2/k?") != std::string::npos) {
      channel = "kline_rest";
    }
  } else {
    LOG(INFO) << "[Bit Feed Processor] meta msg for " << packet.topic_name
              << ": " << packet.payload.data();
    upd_mgr_.onEndPacket(packet);
    return;
  }

  if (!CheckChannelFilter(channel)) {
    upd_mgr_.onEndPacket(packet);
    return;
  }

  if (channel == "book") {
    std::string_view native_symbol = document["M"].GetString();
    auto* book = upd_mgr_.GetBookMutable(native_symbol);
    if (book) {
      if (MaicoinFeedUpdateParser::ParseBook(book, document)) {
        upd_mgr_.PublishBook();
      }
    }
  } else if (channel == "trade") {
    std::string_view native_symbol = document["M"].GetString();
    if (!upd_mgr_.is_symbol_registered(native_symbol)) {
      upd_mgr_.onEndPacket(packet);
      return;
    }

    std::string_view event = document["e"].GetString();
    if (event == "snapshot") return;

    int last =  document["t"].GetArray().Size();
    int index = 1;
    for (const auto& trade_json : document["t"].GetArray()) {
      auto* trade = upd_mgr_.GetTradeMutable();
      MaicoinFeedUpdateParser::ParseTrade(trade_json, trade);
      upd_mgr_.PublishTrade(native_symbol, index != last);
      index++;
    }
  } else if (channel == "kline_rest") {
    std::string_view url = document["url"].GetString();
    auto parameters = base::feed::FeedParsingProcessor::GetRequestParameters(url);
    auto interval = MaicoinFeedUpdateParser::ParseKlineInterval(parameters["period"]);
    const std::string& native_symbol = parameters["market"];
    auto* klines = upd_mgr_.GetKlineMutable(native_symbol, interval);
    if (klines != nullptr) {
      const auto& data = document["data"].GetArray();
      std::optional<int64_t> first_open_timestamp;
      if (data.Size() > 0) {
        first_open_timestamp = stoll(data[0][0].GetString()) * 1'000'000'000LL;
      }

      if (klines->Size() == 0) {
        klines->SnapshotClear();
      } else if (first_open_timestamp &&
                 first_open_timestamp < klines->RBegin()->open_timestamp) {
        klines->SnapshotClear();
      } else {
        klines->UpdateClear();
      }
      bool to_publish = false;
      for (const auto& kline_data : data) {
        bool freezed = MaicoinFeedUpdateParser::ParseKlineRest(kline_data,
            klines, interval, packet.timestamp);
        to_publish = to_publish || freezed;
      }
      if (to_publish) upd_mgr_.PublishKline(native_symbol);
    }
  } else {
    LOG_FIRST_N(INFO, 100) << "[Maicoin Feed Processor] meta msg for " << packet.topic_name
              << ": " << packet.payload.data();
  }

  upd_mgr_.onEndPacket(packet);
}

}  // namespace coin2::exchange::maicoin::feed_v2::impl
